package study.zhangbo.lock.distributed.zk;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import study.zhangbo.lock.distributed.IBaseLockerService;

import java.io.IOException;
import java.util.Collections;
import java.util.List;

@Slf4j
@Component
public class ZookeeperLockService implements IBaseLockerService {

    private static final ThreadLocal<ZooKeeper> ZOO_KEEPER_THREAD_LOCAL = new ThreadLocal<>();

    private static final ThreadLocal<String> LOCK_PATH = new ThreadLocal<>();

    @Value("${zk.address:127.0.0.1:2181}")
    private String zkAddress;

    @Value("${zk.timeout:5000}")
    private int timeout;

    @Value("${lock.root.path:/Locks}")
    private String lockRootPath;

    public static final String LOCK_NODE_NAME = "Lock_";

    private Watcher watch = new Watcher() {
        @Override
        public void process(WatchedEvent watchedEvent) {
            log.debug("{}前锁释放", watchedEvent.getPath());
            synchronized (this) {
                notifyAll();
            }
        }
    };

    @Override
    public void acquireLock() throws IOException {
        initClient();
        try {
            createLock();
            attempLock();
        } catch (KeeperException e) {
            log.error("创建锁异常：{}", e);
            remove();
        } catch (InterruptedException e) {
            log.error("创建锁中断异常：{}", e);
            remove();
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 初始化构建锁
     *
     * @throws IOException
     */
    private void initClient() throws IOException {
        remove();
        ZooKeeper zkClient = new ZooKeeper(zkAddress, timeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                log.debug("event state:{}", watchedEvent.getState());
            }
        });
        try {
            Stat exists = zkClient.exists(lockRootPath, false);
            if (exists == null) {
                createRootPath(zkClient);
            }
        } catch (KeeperException e) {
            log.error("创建根节点异常：{}", e);
        } catch (InterruptedException e) {
            log.error("创建根节点异常：{}", e);
            Thread.currentThread().interrupt();
        }
        ZOO_KEEPER_THREAD_LOCAL.set(zkClient);
    }


    /**
     * 创建锁节点
     */
    private void createLock() throws KeeperException, InterruptedException {
        /*校验根节点是否存在，不存在则创建*/
        ZooKeeper zkClient = ZOO_KEEPER_THREAD_LOCAL.get();
        /*创建临时顺序编号节点*/
        String lockPath = zkClient.create(lockRootPath + "/" + LOCK_NODE_NAME, Thread.currentThread().getName().getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL_SEQUENTIAL);
        log.debug("线程：{},创建销成功 ：{}", Thread.currentThread().getName(), lockPath);
        LOCK_PATH.set(lockPath);
    }

    private synchronized void createRootPath(ZooKeeper zkClient) throws KeeperException, InterruptedException {
        log.info("创建分布式锁zk根节点");
        Stat exists = zkClient.exists(lockRootPath, false);
        if (exists == null) {
            zkClient.create(lockRootPath, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    /**
     * 尝试获取锁
     */
    private void attempLock() throws KeeperException, InterruptedException {
        ZooKeeper zkClient = ZOO_KEEPER_THREAD_LOCAL.get();
        String lockPath = LOCK_PATH.get();
        List<String> lockpaths = zkClient.getChildren(lockRootPath, false);
        Collections.sort(lockpaths);
        int index = lockpaths.indexOf(lockPath.substring(lockRootPath.length() + 1));
        /*校验是否能直接获取到锁，如果为最小的节点则直接获取到锁*/
        if (index == 0) {
            log.debug("线程：{},获取到锁:{}", Thread.currentThread().getName(), lockPath);
            return;
        } else {
            /*如果不为最小节点，则监听前置节点*/
            String preLockPath = lockpaths.get(index - 1);
            Stat stat = zkClient.exists(lockRootPath + "/" + preLockPath, watch);

            if (stat == null) {
                /*如果前一个节点不存在了，则重新获取锁*/
                attempLock();
            } else {
                log.debug("等待前锁释放,preLockPath:{}", preLockPath);
                synchronized (watch) {
                    watch.wait();
                }
                attempLock();
            }
        }
    }

    /**
     * 释放锁
     */
    @Override
    public void releaseLock() {
        ZooKeeper zkClient = ZOO_KEEPER_THREAD_LOCAL.get();
        String lockPath = LOCK_PATH.get();
        try {
            zkClient.delete(lockPath, -1);
            zkClient.close();
            log.info("释放path:{}", lockPath);
        } catch (InterruptedException e) {
            log.error("释放锁时中断异常：{}", e);
            Thread.currentThread().interrupt();
        } catch (KeeperException e) {
            log.error("释放销异常：{}", e);
        } finally {
            remove();
        }
    }

    /**
     * 清空ThreadLocal对象
     */
    private void remove() {
        ZOO_KEEPER_THREAD_LOCAL.remove();
        LOCK_PATH.remove();
    }
}